package com.lagou.flume;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.compress.utils.Charsets;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.event.SimpleEvent;
import org.apache.flume.interceptor.Interceptor;
import org.junit.Test;

import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class WylIntercepter implements Interceptor {
    @Override
    public void initialize() {

    }

    @Override
    public Event intercept(Event event) {
        //获取event body
        String eventBody = new String(event.getBody(), Charsets.UTF_8);
        //获取 event header
        Map<String, String> headersMap = event.getHeaders();
        //解析body 获取json
        String[] bodyArr = eventBody.split("\\s+");
        try {
            String jsonStr = bodyArr[6];
            // 解析json串获取时间戳
            JSONObject jsonObject = JSON.parseObject(jsonStr);
            String timestampStr = jsonObject.getJSONObject("app_active").getString("time");
            // 将时间戳转换为字符串 "yyyy-MM-dd"
            // 将字符串转换为Long
            long timestamp = Long.parseLong(timestampStr);
            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");
            Instant instant = Instant.ofEpochMilli(timestamp);
            LocalDateTime localDateTime = LocalDateTime.ofInstant(instant, ZoneId.systemDefault());
            String date = formatter.format(localDateTime);
            // 将转换后的字符串放置header中
            headersMap.put("logtime", date);
            event.setHeaders(headersMap);
        } catch (Exception e) {
            headersMap.put("logtime", "Unknown");
            event.setHeaders(headersMap);
        }
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> list) {
        List<Event> lstEvent = new ArrayList<>();
        for (Event event : list) {
            Event intercept = intercept(event);
            if (intercept != null) {
                lstEvent.add(intercept);
            }
        }
        return lstEvent;
    }

    @Override
    public void close() {

    }

    public static class Builder implements Interceptor.Builder {
        private static WylIntercepter wylIntercepter;

        static {
            wylIntercepter = new WylIntercepter();
        }

        @Override
        public Interceptor build() {
            //使用单例优化
            return wylIntercepter;
        }

        @Override
        public void configure(Context context) {

        }
    }

    @Test
    public void testJunit() {
        String str = "2020-08-20 11:56:08.211 [main] INFO  com.lagou.ecommerce.AppStart - {'app_active':{'name':'app_active','json':{'entry':'2','action':'0','error_code':'0'},'time':1595288248066},'attr':{'area':'三门峡','uid':'2F10092A1','app_v':'1.1.0','event_type':'common','device_id':'1FB872-9A1001','os_type':'0.97','channel':'WM','language':'chinese','brand':'xiaomi-3'}}";
        Map<String, String> map = new HashMap<>();
        // new Event
        Event event = new SimpleEvent();
        event.setHeaders(map);
        event.setBody(str.getBytes(Charsets.UTF_8));
        // 调用interceptor处理event
        Builder builder = new Builder();
        Interceptor interceptor = builder.build();
        Event outEvent = interceptor.intercept(event);

        // 处理结果
        Map<String, String> headersMap = outEvent.getHeaders();
        System.out.println(JSON.toJSONString(headersMap));
    }
}
